package com.chis.kafka;

import com.alibaba.fastjson.JSON;
import com.chis.jm.bean.TriggerBean;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Arrays;
import java.util.Properties;

/**
 * 获取kafka消息
 * @author wlj
 * @version 2019年03月26日
 * @Description
 * @Company zwx
 */
public class JmKafkaInfo {

    public static void main(String[] args) {
        String kafkaurl = "yfjz-209:9092";
        String topicname = "TOPIC_TRACE";
        Integer part = 3;
        Long offset = 5500L;

        Properties props = new Properties();
        //用于初始化时建立链接到kafka集群
        props.put("bootstrap.servers", kafkaurl);
//        props.put("bootstrap.servers", "10.41.4.177:7113,10.41.4.178:7113,10.41.4.179:7113,10.41.4.180:7113,10.41.4.181:7113,10.41.4.182:7113,10.41.4.183:7113,10.41.4.184:7113");
//        props.put("bootstrap.servers", "192.168.150.90:9092,192.168.150.91:9092,192.168.150.92:9092");
        //kafka使用消费者分组的概念来允许多个消费者共同消费和处理同一个topic中的消息。分组中消费者成员是动态维护的，如果一个消费者处理失败了，那么之前分配给它的partition将被重新分配给分组中其他消费者；同样，如果分组中加入了新的消费者，也将触发整个partition的重新分配，每个消费者将尽可能的分配到相同数目的partition，以达到新的均衡状态；
        props.put("group.id", "testkafka");
        //用于配置是否自动的提交消费进度
        props.put("enable.auto.commit", "true");
        //用于配置自动提交消费进度的时间
        props.put("auto.commit.interval.ms", "1000");
        //会话超时时长，客户端需要周期性的发送“心跳”到broker，这样broker端就可以判断消费者的状态，如果消费在会话周期时长内未发送心跳，那么该消费者将被判定为dead，那么它之前所消费的partition将会被重新的分配给其他存活的消费者
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer", StringDeserializer.class);
        props.put("value.deserializer", DecodeingKafka.class);
//        props.put("max.poll.records", 100);
        //安全认证
        KafkaConsumer<String, Object> consumer = new KafkaConsumer<>(props);

        consumer.assign(Arrays.asList(new TopicPartition("TOPIC_TRACE", 3)));
        //不改变当前offset，指定从这个topic和partition的开始位置获取。
//        consumer.seekToBeginning(Arrays.asList(new TopicPartition(topicName, 1)));
        consumer.seek(new TopicPartition("TOPIC_TRACE", 3), 5500L);

        int countnum = 1;
        while(true) {
            ConsumerRecords<String, Object> records = consumer.poll(1000);
            if(records.isEmpty()){
                break;
            }
            for (ConsumerRecord<String, Object> record : records) {
                TriggerBean triggerBean = (TriggerBean) record.value();
                System.out.println("================="+countnum);
                System.out.println(JSON.toJSONString(triggerBean));
                countnum++;
            }

        }
    }
}
